package com.shujia.mr.kqzl2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class KQZLMapper2 extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        String line = value.toString();

        //1001,万寿西宫,北京
        //20180216-1001 91
        FileSplit inputSplit = (FileSplit) context.getInputSplit(); //获取当前处于哪一个切片
        String fileName = inputSplit.getPath().getName(); //获取数据文件的路径以及名字
        if (fileName.startsWith("city")) {
            String[] strings = line.split(",");
            if (!"id".equals(strings[0])) {
                //处理不是标题的数据
                context.write(new Text(strings[0]), new Text("$" + strings[2] + "-" + strings[1]));
            }
        } else if (fileName.startsWith("part")) {
            String[] split1 = line.split("\t");
            String dateAndId = split1[0]; // 20180216-1001
            String pm25 = split1[1]; // 91
            String[] split2 = dateAndId.split("-");
            String date = split2[0]; // 20180216
            String id = split2[1]; // id
            context.write(new Text(id), new Text("#" + date + "-" + pm25));
        }
    }
}
